package kassiopeia.plugin.mock;

import java.util.concurrent.TimeUnit;

import kassiopeia.plugin.Event;
import kassiopeia.plugin.ProcessorPlugin;

import org.apache.log4j.Logger;

public class MockPlugin extends ProcessorPlugin {

	private Logger log = Logger.getLogger(MockPlugin.class);
	
	private static final String SLEEP_PERIOD_PARAMETER_NAME = "sleep-period";
	private static final String SEND_TO_OUTPUT_PARAMETER_NAME = "send-to-output";
	
	private int SLEEP_PERIOD;
	private boolean SEND_TO_OUTPUT;
	
	private String OUTPUT = "out";
	
	private static volatile long start;
	private static volatile long eventCounter = 0L;
	private static volatile long freezedEventCounter = 0L;
	
	@Override
	public void beforeTask() {
		log.info("Configuration for MockPlugin " + getContext().getPluginInstanceId() + " started");
		
		try {
			SLEEP_PERIOD = Integer.parseInt(getContext().getParameter(SLEEP_PERIOD_PARAMETER_NAME));
			SEND_TO_OUTPUT = Boolean.parseBoolean(getContext().getParameter(SEND_TO_OUTPUT_PARAMETER_NAME));
		} catch (RuntimeException ex) {
			throw new RuntimeException("Configuration error for MockPlugin " + getContext().getPluginInstanceId(), ex);
		}
		
		log.info("Configuration for MockPlugin " + getContext().getPluginInstanceId() + " finished");
	}

	@Override
	public void afterTask() {
		log.info("MockPlugin " + getContext().getPluginInstanceId() + " finished work");
	}

	@Override
	public void process(Event event) {
		event.setBody("MockPlugin:" + getContext().getPluginInstanceId() + ":" + getContext().getThisAgentId());
		
		if (SEND_TO_OUTPUT) {
			sleep();
			send(event);
		} else {
			long tmp = ++eventCounter;
			if (System.currentTimeMillis() - start > 5000L) {
				start = System.currentTimeMillis();
				log.info("THROUGHPUT: " + (tmp - freezedEventCounter)/5);
				freezedEventCounter = tmp;
			}
		}
	}
	
	private void send(Event event) {
		getContext().send(event, OUTPUT);
	}
	
	private void sleep() {
		try {
			TimeUnit.MILLISECONDS.sleep(SLEEP_PERIOD);
		} catch (InterruptedException e) {
			log.warn("Error during sleep for MockPlugin " + getContext().getPluginInstanceId(), e);
		}
	}

}
